package com.haotian.kafka;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.streams.state.HostInfo;

public final class _KafkaConfig {

	private final Properties producer_props = new Properties();
	private final Properties consumer_props = new Properties();

	private List<HostInfo> hosts = new ArrayList<>();

	private List<String> topics = new ArrayList<>();

	/**
	 * 设置kafka服务器地址 格式 ip:端口 集群用,号分割
	 * 
	 * @param hosts
	 */
	public _KafkaConfig(String hosts) {

		String[] tmp = hosts.split(",");

		for (String host : tmp) {
			String[] tmp1 = host.split(":");
			HostInfo hi = new HostInfo(tmp1[0], Integer.parseInt(tmp1[1]));
			this.hosts.add(hi);
		}

		producer_props.put("bootstrap.servers", hosts);
		// 即所有副本都同步到数据时send方法才返回,以此来完全判断数据是否发送成功, 理论上来讲数据不会丢失.
		producer_props.put("acks", "all");
		// 发送失败重试次数
		producer_props.put("retries", 0);
		// 批处理条数：当多个记录被发送到同一个分区时，生产者会尝试将记录合并到更少的请求中。这有助于客户端和服务器的性能。
		producer_props.put("batch.size", 10);
		// 批处理延迟时间上限：即1ms过后，不管是否达到批处理数，都直接发送一次请求
		producer_props.put("linger.ms", 1);
		// 单位：字 节 ，
		// 生产者缓冲大小，当缓冲区耗尽后，额外的发送调用将被阻塞。时间超过max.block.ms将抛出TimeoutException
		producer_props.put("buffer.memory", 33554432);
		// 序列化插件
		producer_props.put("key.serializer", StringSerializer.class.getName());
		producer_props.put("value.serializer", StringSerializer.class.getName());

		consumer_props.put("bootstrap.servers", hosts);
		// 消费者群组ID，发布-订阅模式，即如果一个生产者，多个消费者都要消费，那么需要定义自己的群组，同一群组内的消费者只有一个能消费到消息
		// 修改成全新的没有被读取过的分组id 会重新读取全部topics 的 内容
		consumer_props.put("group.id", "defaultGroup");
		// 如果为true，消费者的偏移量将在后台定期提交。
		consumer_props.put("enable.auto.commit", true);
		// 如果设置为自动提交（enable.auto.commit=true），这里设置自动提交周期
		consumer_props.put("auto.commit.interval.ms", 1);

		// 每次poll的数量限制
		consumer_props.put("max.poll.records", 10);
		// 从何处开始消费,lates 表示消费最新消息,earliest 表示从头开始消费,none表示抛出异常,默认latest
		consumer_props.put("auto.offset.reset", "latest");
		// 反序列化插件
		consumer_props.put("key.deserializer", StringDeserializer.class.getName());
		consumer_props.put("value.deserializer", StringDeserializer.class.getName());// 反序列化插件

	}

	/**
	 * 默认值为：all 即所有副本都同步到数据时send方法才返回, 以此来完全判断数据是否发送成功, 理论上来讲数据不会丢失.
	 * 
	 * @param acks
	 */
	public void setAcks(String acks) {
		producer_props.put("acks", acks);
	}

	/**
	 * 发送失败重试次数
	 * 
	 * @param retries
	 */
	public void setRetries(int retries) {
		producer_props.put("retries", retries);
	}

	/**
	 * 批处理条数：当多个记录被发送到同一个分区时，生产者会尝试将记录合并到更少的请求中。这有助于客户端和服务器的性能。
	 * 
	 * @param batchSize
	 */
	public void setBatchSize(int batchSize) {
		producer_props.put("batch.size", batchSize);
	}

	/**
	 * 批处理延迟时间上限：即1ms过后，不管是否达到批处理数，都直接发送一次请求
	 * 
	 * @param lingerMs
	 */
	public void setLingerMs(int lingerMs) {
		producer_props.put("linger.ms", lingerMs);
	}

	/**
	 * 
	 * 单位：字节
	 * 
	 * 生产者缓冲大小，当缓冲区耗尽后，额外的发送调用将被阻塞。时间超过max.block.ms将抛出TimeoutException
	 * 
	 * @param bufferMemory
	 */
	public void setBufferMemory(long bufferMemory) {
		producer_props.put("buffer.memory", bufferMemory);
	}

	/**
	 * 消费者群组ID，发布-订阅模式，即如果一个生产者，多个消费者都要消费，那么需要定义自己的群组，同一群组内的消费者只有一个能消费到消息
	 * 修改成全新的没有被读取过的分组id 会重新读取全部topics 的 内容
	 * 
	 * @param groupID
	 */
	public void setGroupID(String groupID) {
		consumer_props.put("group.id", groupID);
	}

	/**
	 * 如果为true，消费者的偏移量将在后台定期提交。
	 * 
	 * @param enableAutoCommit
	 */
	public void setEnableAutoCommit(boolean enableAutoCommit) {
		consumer_props.put("enable.auto.commit", enableAutoCommit);
	}

	/**
	 * 如果设置为自动提交（enable.auto.commit=true），这里设置自动提交周期 毫秒
	 * 
	 */
	public void setAutoCommitIntervalMs(long ms) {
		consumer_props.put("auto.commit.interval.ms", ms);
	}

	/**
	 * 从poll(拉)的回话处理时长 毫秒
	 */
	public void setSessionTimeoutMs(long ms) {
		consumer_props.put("session.timeout.ms", ms);
	}

	/**
	 * 每次poll的数量限制
	 */
	public void setMaxPollRecords(int pollMax) {
		consumer_props.put("max.poll.records", pollMax);
	}

	public final String AutoOffsetReset_LATEST = "latest";
	public final String AutoOffsetReset_EARLIEST = "earliest";
	public final String AutoOffsetReset_NONE = "none";

	/**
	 * 从何处开始消费,latest 表示消费最新消息,earliest 表示从头开始消费,none表示抛出异常,默认latest
	 */

	public void setAutoOffsetReset(String aor) {
		consumer_props.put("auto.offset.reset", aor);
	}

	protected List<HostInfo> getHosts() {
		return this.hosts;
	}

	/**
	 * 添加要监听的topic 多此调用即添加多个监听
	 * 
	 * @param topic
	 */
	public void addTopic(String topic) {
		topics.add(topic);
	}

	protected List<String> getTopics() {
		return topics;
	}

	protected Properties getProducer_props() {
		return producer_props;
	}

	protected Properties getConsumer_props() {
		return consumer_props;
	}

}
